package com.atguigu.chapter07.state;

import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/6/15 11:36
 */
public class Flink02_State_Operator_BD {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);
    
        MapStateDescriptor<String, String> bdStateDesc = new MapStateDescriptor<>("bdStateDesc", String.class, String.class);
    
        DataStreamSource<String> dataStream = env
            .socketTextStream("hadoop162", 9999);
    
        BroadcastStream<String> controlStream = env
            .socketTextStream("hadoop162", 8888)
            .broadcast(bdStateDesc);
    
        dataStream
            .connect(controlStream)
            .process(new BroadcastProcessFunction<String, String, String>() {
                @Override
                public void processElement(String value,
                                           ReadOnlyContext ctx,
                                           Collector<String> out) throws Exception {
                    // 处理数据的逻辑
                    ReadOnlyBroadcastState<String, String> bdState = ctx.getBroadcastState(bdStateDesc);
                    String aSwitch = bdState.get("switch");
                    System.out.println(aSwitch);
                    if ("1".equals(aSwitch)) {
                        out.collect("业务逻辑切换 1 号");
                    }else if("2".equals(aSwitch)){
                        out.collect("业务逻辑切换 2 号");
                    }else{
                        out.collect("切换到 默认处理逻辑");
                    }
                }
            
                @Override
                public void processBroadcastElement(String value,
                                                    Context ctx,
                                                    Collector<String> out) throws Exception {
                       // 处理广播元素: 把收到的值, 放入广播状态. 处理处理逻辑中就可以拿到这个广播状态
                    ctx.getBroadcastState(bdStateDesc).put("switch", value);
                    
                }
            })
            .print();
    
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
